package com.ttg.web.service.Impl;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.ttg.common.config.DynamicTableNameHelper;
import com.ttg.common.constant.BasicConstant;
import com.ttg.common.enums.BankFileTypeEnum;
import com.ttg.common.enums.ErrorCodeEnums;
import com.ttg.common.enums.ValueNullAllowEnum;
import com.ttg.common.exception.BaseException;
import com.ttg.common.utils.CommUtils;
import com.ttg.common.utils.DateTimeUtil;
import com.ttg.common.utils.JsonUtils;
import com.ttg.common.utils.QxAndJhUtil;
import com.ttg.model.pojo.*;
import com.ttg.web.dao.*;
import com.ttg.web.service.VipBankInfoOriginalService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.map.CaseInsensitiveMap;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.Period;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * @author zhangHongYuan
 * @date 2023/7/17 11:27
 */
@Service
@Slf4j
public class VipBankInfoOriginalServiceImpl implements VipBankInfoOriginalService {

    /**
     * 线程池最大线程数
     */
    @Value("${processData.nThreads:50}")
    private Integer nThreads;
    /**
     * 队列最大数量
     */
    @Value("${processData.queueSize:100000}")
    private Integer queueSize;
    /**
     * 队列插入最大等待时间 毫秒
     */
    @Value("${processData.queueInsertTimeout:30000}")
    private Integer queueInsertTimeout;
    /**
     * 队列获取最大等待时间 毫秒
     */
    @Value("${processData.queuePullTimeout:5000}")
    private Integer queuePullTimeout;
    /**
     * 线程结束错误计数
     */
    @Value("${processData.maxErrNum:30}")
    private Integer maxErrNum;
    /**
     * 自旋间隔时间 毫秒
     * 每次自旋会增加错误计数，超过则强制退出
     */
    @Value("${processData.spinLocWaitTime:1000}")
    private Integer spinLocWaitTime;


    @Autowired
    private ConditionsNameDao conditionsNameDao;

    @Autowired
    private BankAgentDao bankAgentDao;
//
//    @Autowired
//    private BankServicDao bankServicDao;

    @Autowired
    private BankPublicPrimitiveDao bankPublicPrimitiveDao;

    @Autowired
    private BankCardDataInfoDao bankCardDataInfoDao;

    @Autowired
    private BankProductDataInfoDao bankProductDataInfoDao;

    @Autowired
    private BankCustomerSignInfoDao bankCustomerSignInfoDao;

    @Autowired
    private BankCardBindingInfoDao bankCardBindingInfoDao;

    @Autowired
    private BankPublicTradePrimitiveDao bankPublicTradePrimitiveDao;

    @Autowired
    private VipInfoDao vipInfoDao;

    @Autowired
    private BankPrivatePrimitiveDao bankPrivatePrimitiveDao;

    @Autowired
    private BankPublicPrimitiveAumDao bankPublicPrimitiveAumDao;

    @Autowired
    private DictDetailDao dictDetailDao;

    @Autowired
    private QyxxCustAumDao qyxxCustAumDao;
    @Autowired
    private QyyxCustGroupDao qyyxCustGroupDao;
    @Autowired
    private QyyxCustSignTagDao qyyxCustSignTagDao;
    @Autowired
    private QyyxDepositAccountDetailDao qyyxDepositAccountDetailDao;
    @Autowired
    private QyyxDepositAmountDao qyyxDepositAmountDao;


    /**
     * 数仓推送过来的机构信息-M_UD_GXD_BNK_ORG_F
     *
     * @param filePath 文件路径
     */
    @Override
    public void bankAgent(String filePath) {
        //银行机构表-m_ud_gxd_bnk_org	日	全量 	数据文件	m_ud_gxd_bnk_org.unl
        String fileName = "m_ud_gxd_bnk_org.unl";
        log.info("【VipBankInfoServiceImpl bankAgent】start====>{}", filePath + fileName);
        // 判断下文件是否存在
        File dataFile = new File(CommUtils.filePathFilterClean(filePath + fileName));
        if(!dataFile.exists()){
            log.error("【银行机构表-m_ud_gxd_bnk_org-error】m_ud_gxd_bnk_org 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【银行机构表-m_ud_gxd_bnk_org-error】m_ud_gxd_bnk_org 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        // 先把数据清空,再插入
        bankAgentDao.truncate();

        try {
            List<BankAgent> list = new ArrayList<>();
            BankAgent bankAgent;
            long count = 0L;
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                bankAgent = new BankAgent();
                String[] values = line.split("\u0001");
                if (values.length < 4) {
                    log.info("【银行机构表-m_ud_gxd_bnk_org】 该条数据异常,需剔除 line = {}", line);
                    continue;
                }
                if(StringUtils.isBlank(values[0])){
                    log.info("【银行机构表-m_ud_gxd_bnk_org】 该条数据异常,需剔除 line = {}", line);
                    continue;
                }

                // 机构表	1	ETL_DT	ETL日期
                // 机构表	2	ORG_NO	机构号
                // 机构表	3	ORG_NM	机构名称
                // 机构表	4	F_ORG	上级机构
                // 机构表	5	ORG_LEV	机构级别

                bankAgent.setBankAgentNo(values[0]);
                bankAgent.setBankAgentName(values[1]);
                bankAgent.setBankAgentParentNo(values[2]);
                bankAgent.setBankAgentLevel(values[3]);


                if(StringUtils.isBlank(bankAgent.getBankAgentNo())){
                    log.info("【银行机构表-m_ud_gxd_bnk_org agentNo is null 】 该条数据异常,需剔除 line = {}", line);
                    continue;
                }

                if(StringUtils.isBlank(bankAgent.getBankAgentLevel())){
                    log.info("【银行机构表-m_ud_gxd_bnk_org agentLevel is null】 该条数据异常,需剔除 line = {}", line);
                    continue;
                }



                switch (bankAgent.getBankAgentLevel()) {
                    case "0":
                        bankAgent.setBankAgentNo0(bankAgent.getBankAgentNo());
                        break;
                    case "1":
                        bankAgent.setBankAgentNo1(bankAgent.getBankAgentNo());
                        bankAgent.setBankAgentNo0(bankAgent.getBankAgentParentNo());
                        break;
                    case "2":
                        bankAgent.setBankAgentNo2(bankAgent.getBankAgentNo());
                        bankAgent.setBankAgentNo1(bankAgent.getBankAgentParentNo());
                        break;
                    case "3":
                        bankAgent.setBankAgentNo3(bankAgent.getBankAgentNo());
                        bankAgent.setBankAgentNo2(bankAgent.getBankAgentParentNo());
                        break;
                    case "4":
                        bankAgent.setBankAgentNo4(bankAgent.getBankAgentNo());
                        bankAgent.setBankAgentNo3(bankAgent.getBankAgentParentNo());
                        break;
                }
                // 最后一个字段是更新时间
                bankAgent.setBankAgentUpdateTime(LocalDateTime.now());
                list.add(bankAgent);
                if (BasicConstant.INT_1000 == list.size()) {
                    bankAgentDao.saveBatch(list);
                    count += list.size();
                    list.clear();
                    log.debug("【银行机构表-m_ud_gxd_bnk_org】-" + LocalDateTime.now() + "已完成" + count);
                }
            }
            log.info("【银行机构表-m_ud_gxd_bnk_org】文件数据读取完毕");

            if (list.size() > BasicConstant.INT_0) {
                bankAgentDao.saveBatch(list);
                count += list.size();
                list.clear();
            }
            log.debug("【银行机构表-m_ud_gxd_bnk_org】-" + LocalDateTime.now() + "最终已完成" + count);

            /**
             * 更新机构表各层级机构
             */
            bankAgentDao.updateAgentNoEveryLevel();

        } catch (Exception e) {
            log.error("【银行机构表-m_ud_gxd_bnk_org-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【银行机构表-m_ud_gxd_bnk_org-error】银行数据读取错误");
        }
        log.info("【银行机构表-m_ud_gxd_bnk_org】end====>{}", filePath + fileName);

    }


    /**
     * 银行卡信息-m_ud_gxd_card
     *
     * @param filePath 文件路径
     */
    @Override
    public void cardDataUpdate(String filePath) {
        //银行卡表	日	增量		数据文件	m_ud_gxd_card.unl
        String fileName = "m_ud_gxd_card.unl";
        log.info("【VipBankInfoServiceImpl cardDataUpdate】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【银行卡信息-m_ud_gxd_card】m_ud_gxd_card 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【银行卡信息-m_ud_gxd_card】m_ud_gxd_card 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }

        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_CARD_DATA_INFO.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
//            List<BankCardDataInfo> list = new ArrayList<>();
            Map<String, BankCardDataInfo> batchInsertMap = new CaseInsensitiveMap<>();
            List<Map<String, String>> existsWhere = new ArrayList<>();
            BankCardDataInfo bankCardDataInfo;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【银行卡信息-m_ud_gxd_card】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchInsertMap.values().size());
                            if (batchInsertMap.values().size() > BasicConstant.INT_0) {
                                Map<String, Long> existsMap = bankCardDataInfoDao.existsMap(existsWhere);
                                for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                                    batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                                }
                                bankCardDataInfoDao.saveOrUpdateBatch(batchInsertMap.values());
                                count += batchInsertMap.values().size();
                                batchInsertMap.clear();
                                existsWhere.clear();
                            }
                            log.debug("【银行卡信息-m_ud_gxd_card】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }
                    bankCardDataInfo = new BankCardDataInfo();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【银行卡信息-m_ud_gxd_card】"+Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}",conditionsName.getConditionsValueKey(),value,line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(bankCardDataInfo, conditionsName.getConditionsValueKey(), value);
                            }
                        }

                    }
                    if (ignore) {
                        continue;
                    }
                    Map<String, String> uniqMap = new HashMap<>();
                    uniqMap.put("cstId", bankCardDataInfo.getCstId());
                    uniqMap.put("bankCardNumber", bankCardDataInfo.getBankCardNumber());
                    existsWhere.add(uniqMap);
                    bankCardDataInfo.setCardBin(StringUtils.substring(bankCardDataInfo.getBankCardNumber(), 0, 6));
                    bankCardDataInfo.setUpdateTime(LocalDateTime.now());
                    batchInsertMap.put(bankCardDataInfo.getCstId() + bankCardDataInfo.getBankCardNumber(), bankCardDataInfo);
                    if (BasicConstant.INT_1000 == batchInsertMap.values().size()) {
                        Map<String, Long> existsMap = bankCardDataInfoDao.existsMap(existsWhere);
                        for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                            batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                        }
                        bankCardDataInfoDao.saveOrUpdateBatch(batchInsertMap.values());
                        count += batchInsertMap.values().size();
                        batchInsertMap.clear();
                        existsWhere.clear();
                        log.debug("【银行卡信息-m_ud_gxd_card】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }
            } catch (Exception e) {
                rse.set(e);
                log.error("【银行卡信息-m_ud_gxd_card-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(batchInsertMap.values()), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【银行卡信息-m_ud_gxd_card-error】队列插入异常");
                    throw new BaseException("【银行卡信息-m_ud_gxd_card】队列插入异常");
                }
            }
            log.info("【银行卡信息-m_ud_gxd_card】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【银行卡信息-m_ud_gxd_card】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                log.info("【银行卡信息-m_ud_gxd_card】序列数据尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【银行卡信息-m_ud_gxd_card】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【银行卡信息-m_ud_gxd_card】子线程全部完成");
        } catch (Exception e) {
            log.error("【银行卡信息-m_ud_gxd_card-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【银行卡信息-m_ud_gxd_card-error】银行数据读取错误");

        }

        log.info("【银行卡信息-m_ud_gxd_card】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【银行卡信息-m_ud_gxd_card-error】银行数据读取错误");
        }
    }

    /**
     * 个人客户业务表-m_ud_gxd_indv_cust_busi
     *
     * @param filePath 文件路径
     */
    @Override
    public void productDataInfo(String filePath) {
        //个人客户业务表-m_ud_gxd_indv_cust_busi	日	全量 	数据文件	m_ud_gxd_indv_cust_busi.unl
        String fileName = "m_ud_gxd_indv_cust_busi.unl";
        log.info("【VipBankInfoServiceImpl productDataInfo】start====>{}", filePath + fileName);
        LocalDate dataDate = getDataDateFromFilePath(filePath);
        int daysOfWeek = DateTimeUtil.daysOfWeek(dataDate);
        int allDaysOfWeek = DateTimeUtil.allDaysOfWeek(dataDate);
        int daysOfMonth = DateTimeUtil.daysOfMonth(dataDate);
        int allDaysOfMonth = DateTimeUtil.allDaysOfMonth(dataDate);
        int daysOfQuarter = DateTimeUtil.daysOfQuarter(dataDate);
        int allDaysOfQuarter = DateTimeUtil.allDaysOfQuarter(dataDate);
        int daysOfYear = DateTimeUtil.daysOfYear(dataDate);
        int allDaysOfYear = DateTimeUtil.allDaysOfYear(dataDate);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if (!dataFile.exists()) {
            log.error("【个人客户业务表-m_ud_gxd_indv_cust_busi】m_ud_gxd_indv_cust_busi 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        } else {
            // 文件存在大小是否正常
            if (dataFile.length() == 0) {
                log.error("【个人客户业务表-m_ud_gxd_indv_cust_busi】m_ud_gxd_indv_cust_busi 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_PRODUCT_DATA_INFO.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 先把数据清空,再插入
        bankProductDataInfoDao.truncate();
        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            List<BankProductDataInfo> list = new ArrayList<>();
            BankProductDataInfo bankProductDataInfo;
            long count = 0L;
            try {
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户业务表-m_ud_gxd_indv_cust_busi】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", list.size());
                            if (list.size() > BasicConstant.INT_0) {
                                bankProductDataInfoDao.saveBatch(list);
                                count += list.size();
                            }
                            log.debug("【个人客户业务表-m_ud_gxd_indv_cust_busi】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }
                    bankProductDataInfo = new BankProductDataInfo();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户业务表-m_ud_gxd_indv_cust_busi】"+Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}",conditionsName.getConditionsValueKey(),value,line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(bankProductDataInfo, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    // 最后一个字段是更新时间
                    bankProductDataInfo.setUpdateTime(LocalDate.from(LocalDateTime.now()));
                    // 计算实际日均
                    bankProductDataInfo.setWeeksRealAverageAmount(getRealAverage(bankProductDataInfo.getWeeksAverageAmount(), daysOfWeek, allDaysOfWeek));
                    bankProductDataInfo.setMonthlyRealAmount(getRealAverage(bankProductDataInfo.getMonthlyAmount(), daysOfMonth, allDaysOfMonth));
                    bankProductDataInfo.setQuartertlyRealAmount(getRealAverage(bankProductDataInfo.getQuartertlyAmount(), daysOfQuarter, allDaysOfQuarter));
                    bankProductDataInfo.setYearRealAmount(getRealAverage(bankProductDataInfo.getYearAmount(), daysOfYear, allDaysOfYear));
                    list.add(bankProductDataInfo);
                    if (BasicConstant.INT_1000 == list.size()) {
                        bankProductDataInfoDao.saveBatch(list);
                        count += list.size();
                        list.clear();
                        log.debug("【个人客户业务表-m_ud_gxd_indv_cust_busi】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户业务表-m_ud_gxd_indv_cust_busi-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(list), e);
            }
        });

        for (int i = 0; i < nThreads; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户业务表-m_ud_gxd_indv_cust_busi-error】队列插入异常");
                    throw new BaseException("【个人客户业务表-m_ud_gxd_indv_cust_busi 】队列插入异常");
                }
            }
            log.info("【个人客户业务表-m_ud_gxd_indv_cust_busi】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户业务表-m_ud_gxd_indv_cust_busi】序列数据尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户业务表-m_ud_gxd_indv_cust_busi 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                log.info("【个人客户业务表-m_ud_gxd_indv_cust_busi】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户业务表-m_ud_gxd_indv_cust_busi 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户业务表-m_ud_gxd_indv_cust_busi】子线程全部完成");

        } catch (Exception e) {
            log.error("【个人客户业务表-m_ud_gxd_indv_cust_busi-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户业务表-m_ud_gxd_indv_cust_busi-error】银行数据读取错误");
        }
        log.info("【个人客户业务表-m_ud_gxd_indv_cust_busi】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户业务表-m_ud_gxd_indv_cust_busi-error】银行数据读取错误");

        }
    }

    /**
     * 个人客户签约表-m_ud_gxd_indv_cust_sign
     *
     * @param filePath 文件路径
     */
    @Override
    public void customerSignUpdate(String filePath) {

        //个人客户签约表-m_ud_gxd_indv_cust_sign	日	增量		数据文件	m_ud_gxd_indv_cust_sign.unl
        String fileName = "m_ud_gxd_indv_cust_sign.unl";
        log.info("【VipBankInfoServiceImpl customerSignUpdate】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【个人客户签约表-m_ud_gxd_indv_cust_sign】m_ud_gxd_indv_cust_sign 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【个人客户签约表-m_ud_gxd_indv_cust_sign】m_ud_gxd_indv_cust_sign 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_CUSTOMER_SIGN_INFO.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 获取字典
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        bankCustomerSignInfoDao.truncate();

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            List<BankCustomerSignInfo> list = new ArrayList<>();
            BankCustomerSignInfo bankCustomerSignInfo;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户签约表-m_ud_gxd_indv_cust_sign】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", list.size());
                            if (list.size() > BasicConstant.INT_0) {
                                bankCustomerSignInfoDao.saveOrUpdateBatch(list);
                                count += list.size();
                                list.clear();
                            }
                            log.debug("【个人客户签约表-m_ud_gxd_indv_cust_sign】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    bankCustomerSignInfo = new BankCustomerSignInfo();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户签约表-m_ud_gxd_indv_cust_sign】"+Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}",conditionsName.getConditionsValueKey(),value,line);
                                    }
                                    continue;
                                }

                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(bankCustomerSignInfo, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    list.add(bankCustomerSignInfo);
                    if (BasicConstant.INT_1000 == list.size()) {
                        bankCustomerSignInfoDao.saveOrUpdateBatch(list);
                        count += list.size();
                        list.clear();
                        log.debug("【个人客户签约表-m_ud_gxd_indv_cust_sign】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }
            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户签约表-m_ud_gxd_indv_cust_sign-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(list), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户签约表-m_ud_gxd_indv_cust_sign-error】队列插入异常");
                    throw new BaseException("【个人客户签约表-m_ud_gxd_indv_cust_sign 】队列插入异常");
                }
            }
            log.info("【个人客户签约表-m_ud_gxd_indv_cust_sign】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户签约表-m_ud_gxd_indv_cust_sign】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户签约表-m_ud_gxd_indv_cust_sign】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户签约表-m_ud_gxd_indv_cust_sign 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户签约表-m_ud_gxd_indv_cust_sign】子线程全部完成");
        } catch (Exception e) {
            log.error("【个人客户签约表-m_ud_gxd_indv_cust_sign-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户签约表-m_ud_gxd_indv_cust_sign-error】银行数据读取错误");
        }
        log.info("【个人客户签约表-m_ud_gxd_indv_cust_sign】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户签约表-m_ud_gxd_indv_cust_sign-error】银行数据读取错误");
        }
    }

    /**
     * 商户表-m_ud_gxd_mrch
     *
     * @param filePath 文件路径
     */
    @Override
    public void merchantInfoUpdate(String filePath) {

        //商户表-m_ud_gxd_mrch	日	全量		数据文件	m_ud_gxd_mrch.unl 数据是t-2
        String fileName = "m_ud_gxd_mrch.unl";
        log.info("【商户表-m_ud_gxd_mrch】start====>{}", filePath + fileName);
        LocalDate dataDate = getDataDateFromFilePath(filePath).minusDays(1);
        int daysOfMonth = DateTimeUtil.daysOfMonth(dataDate);
        int allDaysOfMonth = DateTimeUtil.allDaysOfMonth(dataDate);
        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if (!dataFile.exists()) {
            log.error("【商户表-m_ud_gxd_mrch】m_ud_gxd_mrch 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR.getErrcode(), "【商户表-m_ud_gxd_mrch】m_ud_gxd_mrch 文件不存在");
        } else {
            // 文件存在大小是否正常
            if (dataFile.length() == 0) {
                log.error("【商户表-m_ud_gxd_mrch】m_ud_gxd_mrch 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_PUBLIC_PRIMITIVE.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 由于 m_ud_gxd_mrch 是全量数据 但是记录了上次成长值所以不能清空
//        bankPublicPrimitiveDao.truncate();

        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
//            List<BankPublicPrimitive> list = new ArrayList<>();
            Map<String, BankPublicPrimitive> batchInsertMap = new CaseInsensitiveMap<>();
            List<String> existsWhere = new ArrayList<>();
            BankPublicPrimitive bankPublicPrimitive;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【商户表-m_ud_gxd_mrch】" + Thread.currentThread().getName() + "- 线程池已关闭，处理完当前子线程剩余任务数{}后，结束当前子线程", batchInsertMap.values().size());
                            if (batchInsertMap.values().size() > BasicConstant.INT_0) {
                                Map<String, Long> existsMap = bankPublicPrimitiveDao.existsMapByMctNo(existsWhere);
                                for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                                    batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                                    batchInsertMap.get(exists.getKey()).setMctGrade(null);
                                    batchInsertMap.get(exists.getKey()).setMctRegType(null);
//                                    batchInsertMap.get(exists.getKey()).setMctAgoGrow(null);

                                }
                                bankPublicPrimitiveDao.saveOrUpdateBatch(batchInsertMap.values());
                                count += batchInsertMap.values().size();
                                existsWhere.clear();
                                batchInsertMap.clear();
                            }
                            log.debug("【商户表-m_ud_gxd_mrch】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    bankPublicPrimitive = new BankPublicPrimitive();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【商户表-m_ud_gxd_mrch】"+Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}",conditionsName.getConditionsValueKey(),value,line);
                                    }
                                    continue;
                                }

                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(bankPublicPrimitive, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }


                    bankPublicPrimitive.setMctClsId(StringUtils.substringBefore(bankPublicPrimitive.getMctIndustry(), ","));


                    if (Objects.nonNull(bankPublicPrimitive.getMctRegTime())) {
                        try {
                            Period period = Period.between(bankPublicPrimitive.getMctRegTime(), LocalDate.now());
                            bankPublicPrimitive.setMctRegisterYears(period.getYears());
                        } catch (Exception ignored) {
                        }
                    }
                    if (Objects.nonNull(bankPublicPrimitive.getMctPractiseTime())) {
                        try {
                            Period period = Period.between(bankPublicPrimitive.getMctPractiseTime(), LocalDate.now());
                            bankPublicPrimitive.setMctPractiseYears(period.getYears());
                        } catch (Exception ignored) {
                        }
                    }


                    bankPublicPrimitive.setMctGrade(0);
                    bankPublicPrimitive.setMctRegType(0);
//                    bankPublicPrimitive.setMctAgoGrow(0L);
                    bankPublicPrimitive.setUpdateTime(LocalDate.from(LocalDateTime.now()));
                    //计算实际均值
                    bankPublicPrimitive.setMctRealSettlementBankAccountMonthAverage(getRealAverage(bankPublicPrimitive.getMctSettlementBankAccountMonthAverage(), daysOfMonth, allDaysOfMonth));
                    ;
                    existsWhere.add(bankPublicPrimitive.getMctNo());
                    batchInsertMap.put(bankPublicPrimitive.getMctNo(), bankPublicPrimitive);
//                    list.add(bankPublicPrimitive);
                    if (BasicConstant.INT_1000 == batchInsertMap.values().size()) {
                        Map<String, Long> existsMap = bankPublicPrimitiveDao.existsMapByMctNo(existsWhere);
                        for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                            batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                            batchInsertMap.get(exists.getKey()).setMctGrade(null);
                            batchInsertMap.get(exists.getKey()).setMctRegType(null);
//                            batchInsertMap.get(exists.getKey()).setMctAgoGrow(null);

                        }
                        bankPublicPrimitiveDao.saveOrUpdateBatch(batchInsertMap.values());
                        count += batchInsertMap.values().size();
                        batchInsertMap.clear();
                        existsWhere.clear();
                        log.debug("【商户表-m_ud_gxd_mrch】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【商户表-m_ud_gxd_mrch-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(batchInsertMap.values()), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【商户表-m_ud_gxd_mrch-error】队列插入异常");
                    throw new BaseException("【商户表-m_ud_gxd_mrch 】队列插入异常");
                }
            }
            log.info("【商户表-m_ud_gxd_mrch】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【商户表-m_ud_gxd_mrch】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【商户表-m_ud_gxd_mrch 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【商户表-m_ud_gxd_mrch 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【商户表-m_ud_gxd_mrch】子线程全部完成");
        } catch (Exception e) {
            log.error("【商户表-m_ud_gxd_mrch-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【商户表-m_ud_gxd_mrch-error】银行数据读取错误");
        }

        log.info("【商户表-m_ud_gxd_mrch】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【商户表-m_ud_gxd_mrch-error】银行数据读取错误");
        }
    }

    /**
     * 权益客户信息-QYYX_CUSTOM_INFO
     *
     * @param filePath 文件路径
     */
    @Override
    public void vipInfoUpdate(String filePath) {
        //个人客户信息表	日	增量		数据文件	QYYX_CUSTOM_INFO.unl
        String fileName = "QYYX_CUSTOM_INFO.unl";
        log.info("【权益客户信息-QYYX_CUSTOM_INFO】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【权益客户信息-QYYX_CUSTOM_INFO-error】QYYX_CUSTOM_INFO 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【权益客户信息-QYYX_CUSTOM_INFO-error】QYYX_CUSTOM_INFO 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_CUSTOMER_INFO.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 取数据字典
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line = null;
//            List<VipInfo> list = new ArrayList<>();
            Map<String, VipInfo> batchInsertMap = new CaseInsensitiveMap<>();
            List<String> existsWhere = new ArrayList<>();
            VipInfo vipInfo;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【权益客户信息-QYYX_CUSTOM_INFO】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchInsertMap.values().size());
                            if (batchInsertMap.values().size() > BasicConstant.INT_0) {
                                List<VipInfo> updateList = new ArrayList<>();
                                Map<String, Long> existsMap = vipInfoDao.existsMapByCtsScId(existsWhere);
                                for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                                    VipInfo tempVipInfo = batchInsertMap.get(exists.getKey());
                                    tempVipInfo.setVipId(exists.getValue());
                                    tempVipInfo.setVipGrade(null);
                                    tempVipInfo.setVipRegTime(null);
                                    tempVipInfo.setVipRegType(null);
//                                    tempVipInfo.setVipAgoGrow(null);
                                    updateList.add(tempVipInfo);
                                    batchInsertMap.remove(exists.getKey());
                                }
                                vipInfoDao.saveBatch(batchInsertMap.values());
                                vipInfoDao.updateBatchById(updateList);
                                count += batchInsertMap.values().size();
                                count += updateList.size();
                                batchInsertMap.clear();
                                existsWhere.clear();
                            }
                            log.debug("【权益客户信息-QYYX_CUSTOM_INFO】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    vipInfo = new VipInfo();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【权益客户信息-QYYX_CUSTOM_INFO】"+Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}",conditionsName.getConditionsValueKey(),value,line);
                                    }
                                    continue;
                                }

                                if ("vipAge".equals(conditionsName.getConditionsValueKey())) {
                                    if (StringUtils.isNotBlank(value)) {
                                        if (value.length() != BasicConstant.INT_8) {
                                            continue;
                                        }
                                        LocalDate localDate = null;
                                        try {
                                            localDate = DateTimeUtil.parseDateFormatStr(value);
                                        } catch (Exception e) {
                                            log.error("【VipBankInfoServiceImpl vipInfoUpdate-dataerror】localDtate===>{}", value, e);
                                            continue;
                                        }
                                        Period period = Period.between(localDate, LocalDate.now());
                                        value = "" + period.getYears();
                                    }
                                }

                                if ("vipSex".equals(conditionsName.getConditionsValueKey())) {
                                    //目前的性别取值为空串，1,2三种情况, 生产如有其他取值怎么处理
                                    //	数仓回复：custom_info文件，性别取值为1,2以外的值，统一视为 0 未知
                                    if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                        value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, "0");
                                    }
                                }else{
                                    if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                        value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                    }
                                }


                                setValue(vipInfo, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }


                    vipInfo.setVipGrade(Double.valueOf(0));
                    vipInfo.setVipRegTime(LocalDateTime.now());
                    vipInfo.setVipRegType(0);
//                    vipInfo.setVipAgoGrow(0L);
                    vipInfo.setUpdateTime(LocalDateTime.now());
                    existsWhere.add(vipInfo.getVipCstScId());
                    batchInsertMap.put(vipInfo.getVipCstScId(), vipInfo);
                    if (BasicConstant.INT_1000 == batchInsertMap.values().size()) {
                        List<VipInfo> updateList = new ArrayList<>();
                        Map<String, Long> existsMap = vipInfoDao.existsMapByCtsScId(existsWhere);
                        for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                            VipInfo tempVipInfo = batchInsertMap.get(exists.getKey());
                            tempVipInfo.setVipId(exists.getValue());
                            tempVipInfo.setVipGrade(null);
                            tempVipInfo.setVipRegTime(null);
                            tempVipInfo.setVipRegType(null);
//                            tempVipInfo.setVipAgoGrow(null);
                            updateList.add(tempVipInfo);
                            batchInsertMap.remove(exists.getKey());

                        }
                        vipInfoDao.saveBatch(batchInsertMap.values());
                        vipInfoDao.updateBatchById(updateList);
                        count += batchInsertMap.values().size();
                        count += updateList.size();
                        batchInsertMap.clear();
                        existsWhere.clear();
                        log.debug("【权益客户信息-QYYX_CUSTOM_INFO】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【权益客户信息-QYYX_CUSTOM_INFO-error】" + Thread.currentThread().getName() + " line = {}", JsonUtils.objectToJson(batchInsertMap.values()), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【权益客户信息-QYYX_CUSTOM_INFO-error】队列插入异常");
                    throw new BaseException("【权益客户信息-QYYX_CUSTOM_INFO 】队列插入异常");
                }
            }
            log.info("【权益客户信息-QYYX_CUSTOM_INFO】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【权益客户信息-QYYX_CUSTOM_INFO】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【权益客户信息-QYYX_CUSTOM_INFO 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【权益客户信息-QYYX_CUSTOM_INFO 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【权益客户信息-QYYX_CUSTOM_INFO】子线程全部完成");
        } catch (Exception e) {
            log.error("【权益客户信息-QYYX_CUSTOM_INFO-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【权益客户信息-QYYX_CUSTOM_INFO-error】银行数据读取错误");

        }
        log.info("【权益客户信息-QYYX_CUSTOM_INFO】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【权益客户信息-QYYX_CUSTOM_INFO-error】银行数据读取错误");
        }
    }

    /**
     * 快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign
     *
     * @param filePath 文件路径
     */
    @Override
    public void bankCardBindingInfoUpdate(String filePath) {
        //  快捷支付签约表	日	增量		数据文件	m_ud_gxd_indv_qkpy_sign.unl
        String fileName = "m_ud_gxd_indv_qkpy_sign.unl";
        log.info("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign-error】m_ud_gxd_indv_qkpy_sign 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign-error】m_ud_gxd_indv_qkpy_sign 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_CARD_BINDING_INFO.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
//            List<BankCardBindingInfo> list = new ArrayList<>();
            Map<String, BankCardBindingInfo> batchInsertMap = new CaseInsensitiveMap<>();
            List<Map<String, String>> existsWhere = new ArrayList<>();
            BankCardBindingInfo bankCardBindingInfo;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchInsertMap.values().size());
                            if (batchInsertMap.values().size() > BasicConstant.INT_0) {
                                Map<String, Long> existsMap = bankCardBindingInfoDao.existsMapByBankNoAndProduct(existsWhere);
                                for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                                    batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                                }
                                bankCardBindingInfoDao.saveOrUpdateBatch(batchInsertMap.values());
                                count += batchInsertMap.values().size();
                                batchInsertMap.clear();
                                existsWhere.clear();
                            }
                            log.debug("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    bankCardBindingInfo = new BankCardBindingInfo();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】"+Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}",conditionsName.getConditionsValueKey(),value,line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(bankCardBindingInfo, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    Map<String, String> uniqMap = new HashMap<>();
                    uniqMap.put("bankCardNumber", bankCardBindingInfo.getBankCardNumber());
                    uniqMap.put("bindingCardProduct", bankCardBindingInfo.getBindingCardProduct());
                    existsWhere.add(uniqMap);
                    bankCardBindingInfo.setUpdateTime(LocalDateTime.now());
                    batchInsertMap.put(bankCardBindingInfo.getBankCardNumber() + bankCardBindingInfo.getBindingCardProduct(), bankCardBindingInfo);
//                    list.add(bankCardBindingInfo);
                    if (BasicConstant.INT_1000 == batchInsertMap.values().size()) {
                        Map<String, Long> existsMap = bankCardBindingInfoDao.existsMapByBankNoAndProduct(existsWhere);
                        for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                            batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                        }
                        bankCardBindingInfoDao.saveOrUpdateBatch(batchInsertMap.values());
                        count += batchInsertMap.values().size();
                        batchInsertMap.clear();
                        existsWhere.clear();
                        log.debug("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);

                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(batchInsertMap.values()), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }
        BankCardBindingInfo bankCardBindingInfo = new BankCardBindingInfo();
        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign-error】队列插入异常");
                    throw new BaseException("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign 】队列插入异常");
                }
            }
            log.info("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】子线程全部完成");
        } catch (Exception e) {
            log.error("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign-error】{}", JsonUtils.objectToJson(bankCardBindingInfo), e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign-error】银行数据读取错误");
        }
        log.info("【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【快捷支付卡签约信息-m_ud_gxd_indv_qkpy_sign-error】银行数据读取错误");
        }
    }

    /**
     * 个人客户AUM-m_ud_gxd_indv_cust_aum
     *
     * @param filePath 文件路径
     */
    @Override
    public void bankPrivatePrimitiveUpdate(String filePath) {
        // 个人客户AUM	日	全量			数据文件	m_ud_gxd_indv_cust_aum.unl
        String fileName = "m_ud_gxd_indv_cust_aum.unl";
        log.info("【个人客户AUM-m_ud_gxd_indv_cust_aum】start====>{}", filePath + fileName);
        LocalDate dataDate = getDataDateFromFilePath(filePath);
        int daysOfWeek = DateTimeUtil.daysOfWeek(dataDate);
        int allDaysOfWeek = DateTimeUtil.allDaysOfWeek(dataDate);
        int daysOfMonth = DateTimeUtil.daysOfMonth(dataDate);
        int allDaysOfMonth = DateTimeUtil.allDaysOfMonth(dataDate);
        int daysOfQuarter = DateTimeUtil.daysOfQuarter(dataDate);
        int allDaysOfQuarter = DateTimeUtil.allDaysOfQuarter(dataDate);
        int daysOfYear = DateTimeUtil.daysOfYear(dataDate);
        int allDaysOfYear = DateTimeUtil.allDaysOfYear(dataDate);
        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if (!dataFile.exists()) {
            log.error("【个人客户AUM-m_ud_gxd_indv_cust_aum-error】m_ud_gxd_indv_cust_aum 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        } else {
            // 文件存在大小是否正常
            if (dataFile.length() == 0) {
                log.error("【个人客户AUM-m_ud_gxd_indv_cust_aum-error】m_ud_gxd_indv_cust_aum 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_PRIVATE_PRIMITIVE.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }

        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);
        List<Integer> clearCondIds = conditionsNameByClassId.stream().filter(e -> {
            return 1 == e.getIsSplitTableField();
        }).map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());

        for (Integer clearCondId : clearCondIds) {
            for (Map.Entry<String, String> entry : dictsMap.get(clearCondId).entrySet()) {
                Map<String, String> tableSuffix = DynamicTableNameHelper.getTableSuffix();
                tableSuffix.put("tbl_bank_private_primitive", entry.getKey());
                DynamicTableNameHelper.setTableSuffix(tableSuffix);
                // 由于m_ud_gxd_indv_cust_aum是全量数据，所以在接受数据前，将表清空
                bankPrivatePrimitiveDao.truncate();
            }

        }

        // 由于m_ud_gxd_indv_cust_aum是全量数据，所以在接受数据前，将表清空
        bankPrivatePrimitiveDao.truncate();

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            String splitField = null;
            List<BankPrivatePrimitive> typeList = new ArrayList<>();
            Map<String, List<BankPrivatePrimitive>> batMap = new CaseInsensitiveMap<>();
            BankPrivatePrimitive bankPrivatePrimitive;
            try {
                long batchCount = 0L;
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户AUM-m_ud_gxd_indv_cust_aum】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchCount);
                            if (batchCount > BasicConstant.INT_0) {
                                bankPrivatePrimitiveDao.saveBatchTypeSplit(batMap);
                                count += batchCount;
                                batMap.clear();
                            }
                            log.debug("【个人客户AUM-m_ud_gxd_indv_cust_aum】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    bankPrivatePrimitive = new BankPrivatePrimitive();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户AUM-m_ud_gxd_indv_cust_aum】" + Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}", conditionsName.getConditionsValueKey(), value, line);
                                    }
                                    continue;
                                }
                                // 根据type来拆分表
                                if (1 == conditionsName.getIsSplitTableField()) {
                                    if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                        if (dictsMap.get(conditionsName.getConditionsDictId()).containsKey(value)) {
                                            splitField = value;
                                            typeList = batMap.getOrDefault(splitField, new ArrayList<>());
                                        } else {
                                            splitField = "";
                                            typeList = batMap.getOrDefault(splitField, new ArrayList<>());

                                        }
                                        value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                    }
                                } else {
                                    if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                        value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                    }
                                }


                                setValue(bankPrivatePrimitive, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }

                    bankPrivatePrimitive.setAmountRealWeeksAverage(getRealAverage(bankPrivatePrimitive.getAmountWeeksAverage(), daysOfWeek, allDaysOfWeek));
                    bankPrivatePrimitive.setAmountRealMonthAverage(getRealAverage(bankPrivatePrimitive.getAmountMonthAverage(), daysOfMonth, allDaysOfMonth));
                    bankPrivatePrimitive.setAmountRealQuarterAverage(getRealAverage(bankPrivatePrimitive.getAmountQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    bankPrivatePrimitive.setAmountRealYearAverage(getRealAverage(bankPrivatePrimitive.getAmountYearAverage(), daysOfYear, allDaysOfYear));

                    bankPrivatePrimitive.setAmountRealLastWeeksAddition(getAddition(bankPrivatePrimitive.getAmountRealWeeksAverage(), bankPrivatePrimitive.getAmountLastWeeksAverage()));
                    bankPrivatePrimitive.setAmountRealLastMonthAddition(getAddition(bankPrivatePrimitive.getAmountRealMonthAverage(), bankPrivatePrimitive.getAmountLastMonthAverage()));
                    bankPrivatePrimitive.setAmountRealLastQuarterAddition(getAddition(bankPrivatePrimitive.getAmountRealQuarterAverage(), bankPrivatePrimitive.getAmountLastQuarterAverage()));
                    bankPrivatePrimitive.setAmountRealLastYearAddition(getAddition(bankPrivatePrimitive.getAmountRealYearAverage(), bankPrivatePrimitive.getAmountLastYearAverage()));

                    bankPrivatePrimitive.setUpdateTime(LocalDateTime.now());
                    batchCount++;
                    typeList.add(bankPrivatePrimitive);
                    batMap.put(splitField, typeList);
                    if (BasicConstant.INT_1000 == batchCount) {
                        bankPrivatePrimitiveDao.saveBatchTypeSplit(batMap);
                        count += batchCount;
                        batchCount = 0;
                        batMap.clear();
                        log.debug("【个人客户AUM-m_ud_gxd_indv_cust_aum】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }
            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户AUM-m_ud_gxd_indv_cust_aum-error】" + Thread.currentThread().getName() + " batMap = {}", JsonUtils.objectToJson(batMap), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName), 40 * 1024);
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户AUM-m_ud_gxd_indv_cust_aum-error】队列插入异常");
                    throw new BaseException("【个人客户AUM-m_ud_gxd_indv_cust_aum 】队列插入异常");
                }
            }
            log.info("【个人客户AUM-m_ud_gxd_indv_cust_aum】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，得接数据

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户AUM-m_ud_gxd_indv_cust_aum】子进程尚未处理完毕");

                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户AUM-m_ud_gxd_indv_cust_aum 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户AUM-m_ud_gxd_indv_cust_aum 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户AUM-m_ud_gxd_indv_cust_aum】子线程全部完成");
        } catch (Exception e) {
            log.error("【个人客户AUM-m_ud_gxd_indv_cust_aum-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户AUM-m_ud_gxd_indv_cust_aum-error】银行数据读取错误");
        }
        log.info("【个人客户AUM-m_ud_gxd_indv_cust_aum】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户AUM-m_ud_gxd_indv_cust_aum-error】银行数据读取错误");
        }
    }

    /**
     * 个人客户AUM-QYYX_CUSTOM_AUM
     *
     * @param filePath 文件路径
     */
    @Override
    public void qyyxCustAumUpdate(String filePath) {
        //个人客户AUM-QYYX_CUSTOM_AUM	日	全量 	数据文件	QYYX_CUSTOM_AUM.unl
        String fileName = "QYYX_CUSTOM_AUM.unl";
        log.info("【VipBankInfoServiceImpl qyxxCustAumUpdate】start====>{}", filePath + fileName);
        LocalDate dataDate = getDataDateFromFilePath(filePath);
        int daysOfQuarter = DateTimeUtil.daysOfQuarter(dataDate);
        int allDaysOfQuarter = DateTimeUtil.allDaysOfQuarter(dataDate);
        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if (!dataFile.exists()) {
            log.error("【VipBankInfoServiceImpl qyxxCustAumUpdate-error】QYYX_CUSTOM_AUM 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        } else {
            // 文件存在大小是否正常
            if (dataFile.length() == 0) {
                log.error("【VipBankInfoServiceImpl qyxxCustAumUpdate-error】QYYX_CUSTOM_AUM 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.QYYX_CUSTOM_AUM.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 先把数据清空,再插入
        qyxxCustAumDao.truncate();
        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            List<QyxxCustAum> list = new ArrayList<>();
            QyxxCustAum qyxxCustAum;
            long count = 0L;
            try {
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户AUM-QYYX_CUSTOM_AUM】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", list.size());
                            if (list.size() > BasicConstant.INT_0) {
                                qyxxCustAumDao.saveBatch(list);
                                count += list.size();
                            }
                            log.debug("【个人客户AUM-QYYX_CUSTOM_AUM】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }
                    qyxxCustAum = new QyxxCustAum();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户AUM-QYYX_CUSTOM_AUM】" + Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}", conditionsName.getConditionsValueKey(), value, line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(qyxxCustAum, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    // 最后一个字段是更新时间
                    qyxxCustAum.setUpdateTime(LocalDateTime.now());
                    // 计算实际日均
                    qyxxCustAum.setAumAmountRealQuarterAverage(getRealAverage(qyxxCustAum.getAumAmountQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setDepAmountRealQuarterAverage(getRealAverage(qyxxCustAum.getDepAmountQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setWealthAmountRealQuarterAverage(getRealAverage(qyxxCustAum.getWealthAmountQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setTmDepAmountRealQuarterAverage(getRealAverage(qyxxCustAum.getTmDepAmountQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setDmdDepAmountRealQuarterAverage(getRealAverage(qyxxCustAum.getDmdDepAmountQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setSelfBizFinRealQuarterAverage(getRealAverage(qyxxCustAum.getSelfBizFinQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setCsnFinRealQuarterAverage(getRealAverage(qyxxCustAum.getCsnFinQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setCsnFundRealQuarterAverage(getRealAverage(qyxxCustAum.getCsnFundQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setCsnInsureRealQuarterAverage(getRealAverage(qyxxCustAum.getCsnInsureQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    qyxxCustAum.setCsnMetalRealQuarterAverage(getRealAverage(qyxxCustAum.getCsnMetalQuarterAverage(), daysOfQuarter, allDaysOfQuarter));

                    list.add(qyxxCustAum);
                    if (BasicConstant.INT_1000 == list.size()) {
                        qyxxCustAumDao.saveBatch(list);
                        count += list.size();
                        list.clear();
                        log.debug("【个人客户AUM-QYYX_CUSTOM_AUM】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户AUM-QYYX_CUSTOM_AUM-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(list), e);
            }
        });

        for (int i = 0; i < nThreads; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户AUM-QYYX_CUSTOM_AUM-error】队列插入异常");
                    throw new BaseException("【个人客户AUM-QYYX_CUSTOM_AUM 】队列插入异常");
                }
            }
            log.info("【个人客户AUM-QYYX_CUSTOM_AUM】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户AUM-QYYX_CUSTOM_AUM】序列数据尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户AUM-QYYX_CUSTOM_AUM 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                log.info("【个人客户AUM-QYYX_CUSTOM_AUM】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户AUM-QYYX_CUSTOM_AUM 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户AUM-QYYX_CUSTOM_AUM】子线程全部完成");

        } catch (Exception e) {
            log.error("【个人客户AUM-QYYX_CUSTOM_AUM-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户AUM-QYYX_CUSTOM_AUM-error】银行数据读取错误");
        }
        log.info("【个人客户AUM-QYYX_CUSTOM_AUM】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户AUM-QYYX_CUSTOM_AUM-error】银行数据读取错误");
        }
    }


    /**
     * 个人客户客群表-QYYX_CUSTOM_GROUP
     *
     * @param filePath 文件路径
     */
    @Override
    public void qyyxCustGroupUpdate(String filePath) {
        //  个人客户客群表	月	全量		数据文件	QYYX_CUSTOM_GROUP.unl
        String fileName = "QYYX_CUSTOM_GROUP.unl";
        log.info("【个人客户客群表-QYYX_CUSTOM_GROUP】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【个人客户客群表-QYYX_CUSTOM_GROUP-error】QYYX_CUSTOM_GROUP 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【个人客户客群表-QYYX_CUSTOM_GROUP-error】QYYX_CUSTOM_GROUP 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.QYYX_CUST_GROUP.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 先把数据清空,再插入
        qyyxCustGroupDao.truncate();

        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            List<QyyxCustGroup> list = new ArrayList<>();
//            Map<String, QyyxCustGroup> batchInsertMap = new CaseInsensitiveMap<>();
            QyyxCustGroup qyyxCustGroup;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户客群表-QYYX_CUSTOM_GROUP】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", list.size());
                            if (list.size() > BasicConstant.INT_0) {
                                qyyxCustGroupDao.saveBatch(list);
                                count += list.size();
                                list.clear();
                            }
                            log.debug("【个人客户客群表-QYYX_CUSTOM_GROUP】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    qyyxCustGroup = new QyyxCustGroup();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户客群表-QYYX_CUSTOM_GROUP】" + Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}", conditionsName.getConditionsValueKey(), value, line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(qyyxCustGroup, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    qyyxCustGroup.setUpdateTime(LocalDateTime.now());
                    list.add(qyyxCustGroup);
                    if (BasicConstant.INT_1000 == list.size()) {
                        qyyxCustGroupDao.saveBatch(list);
                        count += list.size();
                        list.clear();
                        log.debug("【个人客户客群表-QYYX_CUSTOM_GROUP】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);

                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户客群表-QYYX_CUSTOM_GROUP-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(list), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }
        BankCardBindingInfo bankCardBindingInfo = new BankCardBindingInfo();
        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户客群表-QYYX_CUSTOM_GROUP-error】队列插入异常");
                    throw new BaseException("【个人客户客群表-QYYX_CUSTOM_GROUP 】队列插入异常");
                }
            }
            log.info("【个人客户客群表-QYYX_CUSTOM_GROUP】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户客群表-QYYX_CUSTOM_GROUP】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户客群表-QYYX_CUSTOM_GROUP 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户客群表-QYYX_CUSTOM_GROUP 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户客群表-QYYX_CUSTOM_GROUP】子线程全部完成");
        } catch (Exception e) {
            log.error("【个人客户客群表-QYYX_CUSTOM_GROUP-error】{}", JsonUtils.objectToJson(bankCardBindingInfo), e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户客群表-QYYX_CUSTOM_GROUP-error】银行数据读取错误");
        }
        log.info("【个人客户客群表-QYYX_CUSTOM_GROUP】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户客群表-QYYX_CUSTOM_GROUP-error】银行数据读取错误");
        }
    }


    /**
     * 个人客户标签表-QYYX_CUSTOM_SIGN
     *
     * @param filePath 文件路径
     */
    @Override
    public void qyyxCustSignTagUpdate(String filePath) {
        //  个人客户标签表	日	增量		数据文件	QYYX_CUSTOM_SIGN.unl
        String fileName = "QYYX_CUSTOM_SIGN.unl";
        log.info("【个人客户标签表-QYYX_CUSTOM_SIGN】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【个人客户标签表-QYYX_CUSTOM_SIGN-error】QYYX_CUSTOM_SIGN 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【个人客户标签表-QYYX_CUSTOM_SIGN-error】QYYX_CUSTOM_SIGN 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.QYYX_CUST_SIGN_TAG.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            Map<String, QyyxCustSignTag> batchInsertMap = new CaseInsensitiveMap<>();
            List<Map<String, String>> existsWhere = new ArrayList<>();
            QyyxCustSignTag qyyxCustSignTag;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户标签表-QYYX_CUSTOM_SIGN】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchInsertMap.values().size());
                            if (batchInsertMap.values().size() > BasicConstant.INT_0) {
                                Map<String, Long> existsMap = qyyxCustSignTagDao.existsMap(existsWhere);
                                for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                                    batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                                }
                                qyyxCustSignTagDao.saveOrUpdateBatch(batchInsertMap.values());
                                count += batchInsertMap.values().size();
                                batchInsertMap.clear();
                                existsWhere.clear();
                            }
                            log.debug("【个人客户标签表-QYYX_CUSTOM_SIGN】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    qyyxCustSignTag = new QyyxCustSignTag();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户标签表-QYYX_CUSTOM_SIGN】" + Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}", conditionsName.getConditionsValueKey(), value, line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(qyyxCustSignTag, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    Map<String, String> uniqMap = new HashMap<>();
                    uniqMap.put("cstScId", qyyxCustSignTag.getCstScId());
                    uniqMap.put("tagNo", qyyxCustSignTag.getTagNo());
                    existsWhere.add(uniqMap);
                    qyyxCustSignTag.setUpdateTime(LocalDateTime.now());
                    batchInsertMap.put(qyyxCustSignTag.getCstScId()+qyyxCustSignTag.getTagNo(), qyyxCustSignTag);
                    if (BasicConstant.INT_1000 == batchInsertMap.values().size()) {
                        Map<String, Long> existsMap = qyyxCustSignTagDao.existsMap(existsWhere);
                        for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                            batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                        }
                        qyyxCustSignTagDao.saveOrUpdateBatch(batchInsertMap.values());
                        count += batchInsertMap.values().size();
                        batchInsertMap.clear();
                        existsWhere.clear();
                        log.debug("【个人客户标签表-QYYX_CUSTOM_SIGN】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);

                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户标签表-QYYX_CUSTOM_SIGN-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(batchInsertMap.values()), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }
        BankCardBindingInfo bankCardBindingInfo = new BankCardBindingInfo();
        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户标签表-QYYX_CUSTOM_SIGN-error】队列插入异常");
                    throw new BaseException("【个人客户标签表-QYYX_CUSTOM_SIGN 】队列插入异常");
                }
            }
            log.info("【个人客户标签表-QYYX_CUSTOM_SIGN】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户标签表-QYYX_CUSTOM_SIGN】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户标签表-QYYX_CUSTOM_SIGN 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户标签表-QYYX_CUSTOM_SIGN 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户标签表-QYYX_CUSTOM_SIGN】子线程全部完成");
        } catch (Exception e) {
            log.error("【个人客户标签表-QYYX_CUSTOM_SIGN-error】{}", JsonUtils.objectToJson(bankCardBindingInfo), e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户标签表-QYYX_CUSTOM_SIGN-error】银行数据读取错误");
        }
        log.info("【个人客户标签表-QYYX_CUSTOM_SIGN】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户标签表-QYYX_CUSTOM_SIGN-error】银行数据读取错误");
        }
    }

    /**
     * 个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL
     *QYYX_DEPOSIT_ACCOUNT_DETAIL
     * @param filePath 文件路径
     */
    @Override
    public void qyyxDepositAccountDetailUpdate(String filePath) {
        //  个人客户存款账户表	日	增量		数据文件	QYYX_DEPOSIT_ACCOUNT_DETAIL.unl
        String fileName = "QYYX_DEPOSIT_ACCOUNT_DETAIL.unl";
        log.info("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL-error】QYYX_DEPOSIT_ACCOUNT_DETAIL 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL-error】QYYX_DEPOSIT_ACCOUNT_DETAIL 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.QYYX_DEPOSIT_ACCOUNT_DETAIL.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
//            List<BankCardBindingInfo> list = new ArrayList<>();
            Map<String, QyyxDepositAccountDetail> batchInsertMap = new CaseInsensitiveMap<>();
            List<String> existsWhere = new ArrayList<>();
            QyyxDepositAccountDetail qyyxDepositAccountDetail;
            try {
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchInsertMap.values().size());
                            if (batchInsertMap.values().size() > BasicConstant.INT_0) {
                                Map<String, Long> existsMap = qyyxDepositAccountDetailDao.existsMap(existsWhere);
                                for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                                    batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                                }
                                qyyxDepositAccountDetailDao.saveOrUpdateBatch(batchInsertMap.values());
                                count += batchInsertMap.values().size();
                                batchInsertMap.clear();
                                existsWhere.clear();
                            }
                            log.debug("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    qyyxDepositAccountDetail = new QyyxDepositAccountDetail();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】" + Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}", conditionsName.getConditionsValueKey(), value, line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(qyyxDepositAccountDetail, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    Map<String, String> uniqMap = new HashMap<>();
//                    uniqMap.put("cstScId", qyyxDepositAccountDetail.getCstScId());
//                    uniqMap.put("accountNo", qyyxDepositAccountDetail.getAccountNo());
//                    uniqMap.put("accountId", qyyxDepositAccountDetail.getAccountId());
                    existsWhere.add(qyyxDepositAccountDetail.getAccountId());
                    qyyxDepositAccountDetail.setUpdateTime(LocalDateTime.now());
                    batchInsertMap.put(qyyxDepositAccountDetail.getAccountId(), qyyxDepositAccountDetail);
                    if (BasicConstant.INT_1000 == batchInsertMap.values().size()) {
                        Map<String, Long> existsMap = qyyxDepositAccountDetailDao.existsMap(existsWhere);
                        for (Map.Entry<String, Long> exists : existsMap.entrySet()) {
                            batchInsertMap.get(exists.getKey()).setId(exists.getValue());
                        }
                        qyyxDepositAccountDetailDao.saveOrUpdateBatch(batchInsertMap.values());
                        count += batchInsertMap.values().size();
                        batchInsertMap.clear();
                        existsWhere.clear();
                        log.debug("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);

                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(batchInsertMap.values()), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }
        BankCardBindingInfo bankCardBindingInfo = new BankCardBindingInfo();
        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL-error】队列插入异常");
                    throw new BaseException("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL 】队列插入异常");
                }
            }
            log.info("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】子线程全部完成");
        } catch (Exception e) {
            log.error("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL-error】{}", JsonUtils.objectToJson(bankCardBindingInfo), e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL-error】银行数据读取错误");
        }
        log.info("【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户存款账户表-QYYX_DEPOSIT_ACCOUNT_DETAIL-error】银行数据读取错误");
        }
    }

    /**
     * 个人客户存款表-QYYX_DEPOSIT_BAL
     * M_UD_QYYX_INDV_CUST_DEP_BAL_M
     * @param filePath 文件路径
     */
    @Override
    public void qyyxDepositAmountUpdate(String filePath) {
        //个人客户存款表-QYYX_DEPOSIT_BAL	日	全量 	数据文件	QYYX_DEPOSIT_BAL.unl
        String fileName = "QYYX_DEPOSIT_BAL.unl";
        log.info("【VipBankInfoServiceImpl qyyxDepositAmountUpdate】start====>{}", filePath + fileName);

        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if(!dataFile.exists()){
            log.error("【VipBankInfoServiceImpl qyyxDepositAmountUpdate-error】QYYX_DEPOSIT_BAL 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        }else{
            // 文件存在大小是否正常
            if(dataFile.length()==0){
                log.error("【VipBankInfoServiceImpl qyyxDepositAmountUpdate-error】QYYX_DEPOSIT_BAL 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.QYYX_DEPOSIT_AMOUNT.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }
        // 先把数据清空,再插入
        qyyxDepositAmountDao.truncate();
        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            List<QyyxDepositAmount> list = new ArrayList<>();
            QyyxDepositAmount qyyxDepositAmount;
            long count = 0L;
            try {
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【个人客户存款表-QYYX_DEPOSIT_BAL】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", list.size());
                            if (list.size() > BasicConstant.INT_0) {
                                qyyxDepositAmountDao.saveBatch(list);
                                count += list.size();
                            }
                            log.debug("【个人客户存款表-QYYX_DEPOSIT_BAL】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }
                    qyyxDepositAmount = new QyyxDepositAmount();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【个人客户存款表-QYYX_DEPOSIT_BAL】" + Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}", conditionsName.getConditionsValueKey(), value, line);
                                    }
                                    continue;
                                }
                                if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                    value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                }
                                setValue(qyyxDepositAmount, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    // 最后一个字段是更新时间
                    qyyxDepositAmount.setUpdateTime(LocalDateTime.now());
                    list.add(qyyxDepositAmount);
                    if (BasicConstant.INT_1000 == list.size()) {
                        qyyxDepositAmountDao.saveBatch(list);
                        count += list.size();
                        list.clear();
                        log.debug("【个人客户存款表-QYYX_DEPOSIT_BAL】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }

            } catch (Exception e) {
                rse.set(e);
                log.error("【个人客户存款表-QYYX_DEPOSIT_BAL-error】" + Thread.currentThread().getName() + " list = {}", JsonUtils.objectToJson(list), e);
            }
        });

        for (int i = 0; i < nThreads; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【个人客户存款表-QYYX_DEPOSIT_BAL-error】队列插入异常");
                    throw new BaseException("【个人客户存款表-QYYX_DEPOSIT_BAL 】队列插入异常");
                }
            }
            log.info("【个人客户存款表-QYYX_DEPOSIT_BAL】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【个人客户存款表-QYYX_DEPOSIT_BAL】序列数据尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户存款表-QYYX_DEPOSIT_BAL 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                log.info("【个人客户存款表-QYYX_DEPOSIT_BAL】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【个人客户存款表-QYYX_DEPOSIT_BAL 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【个人客户存款表-QYYX_DEPOSIT_BAL】子线程全部完成");

        } catch (Exception e) {
            log.error("【个人客户存款表-QYYX_DEPOSIT_BAL-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户存款表-QYYX_DEPOSIT_BAL-error】银行数据读取错误");
        }
        log.info("【个人客户存款表-QYYX_DEPOSIT_BAL】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【个人客户存款表-QYYX_DEPOSIT_BAL-error】银行数据读取错误");
        }
    }

    @Override
    public void bankPublicPrimitiveAum(String filePath) {
// 对公客户AUM	日	全量			数据文件	m_ud_gxd_ent_cust_aum.unl
        String fileName = "m_ud_gxd_ent_cust_aum.unl";
        log.info("【对公客户AUM-m_ud_gxd_ent_cust_aum】start====>{}", filePath + fileName);
        LocalDate dataDate = getDataDateFromFilePath(filePath);
        int daysOfWeek = DateTimeUtil.daysOfWeek(dataDate);
        int allDaysOfWeek = DateTimeUtil.allDaysOfWeek(dataDate);
        int daysOfMonth = DateTimeUtil.daysOfMonth(dataDate);
        int allDaysOfMonth = DateTimeUtil.allDaysOfMonth(dataDate);
        int daysOfQuarter = DateTimeUtil.daysOfQuarter(dataDate);
        int allDaysOfQuarter = DateTimeUtil.allDaysOfQuarter(dataDate);
        int daysOfYear = DateTimeUtil.daysOfYear(dataDate);
        int allDaysOfYear = DateTimeUtil.allDaysOfYear(dataDate);
        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if (!dataFile.exists()) {
            log.error("【对公客户AUM-m_ud_gxd_ent_cust_aum-error】m_ud_gxd_ent_cust_aum 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        } else {
            // 文件存在大小是否正常
            if (dataFile.length() == 0) {
                log.error("【对公客户AUM-m_ud_gxd_ent_cust_aum-error】m_ud_gxd_ent_cust_aum 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }
        List<String> strs= new ArrayList<String>(
                Arrays.asList("01","02","03","04","05","06","07","08","09","10","11","12","13","14","15","16"));
        for (String str : strs) {
            Map<String, String> tableSuffix = DynamicTableNameHelper.getTableSuffix();
            tableSuffix.put("tbl_bank_public_primitive_aum", str);
            DynamicTableNameHelper.setTableSuffix(tableSuffix);
            bankPublicPrimitiveAumDao.truncate();

        }

        // 由于m_ud_gxd_ent_cust_aum是全量数据，所以在接受数据前，将表清空
        bankPublicPrimitiveAumDao.truncate();

        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            String splitField = null;
            List<BankPublicPrimitiveAum> typeList = new ArrayList<>();
            Map<String, List<BankPublicPrimitiveAum>> batMap = new CaseInsensitiveMap<>();
            BankPublicPrimitiveAum bankPublicPrimitiveAum;
            try {
                long batchCount = 0L;
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【对公客户AUM-m_ud_gxd_ent_cust_aum】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchCount);
                            if (batchCount > BasicConstant.INT_0) {
                                bankPublicPrimitiveAumDao.saveBatchTypeSplit(batMap);
                                count += batchCount;
                                batMap.clear();
                            }
                            log.debug("【对公客户AUM-m_ud_gxd_ent_cust_aum】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    bankPublicPrimitiveAum = new BankPublicPrimitiveAum();
                    String[] values = line.split("\u0001");
                    bankPublicPrimitiveAum.setDataTime(values[0]);
                    bankPublicPrimitiveAum.setCoreCustNo(values[1]);
                    bankPublicPrimitiveAum.setDataType(values[2]);

                    bankPublicPrimitiveAum.setCertTyp(values[3]);
                    bankPublicPrimitiveAum.setCertNo(values[4]);
                    bankPublicPrimitiveAum.setCertNm(values[5]);
                    bankPublicPrimitiveAum.setCustSts(values[6]);
                    bankPublicPrimitiveAum.setAmountTime(Objects.isNull(values[7])?null:new BigDecimal(values[7]));
                    bankPublicPrimitiveAum.setAmountYesterdayDailyAverage(Objects.isNull(values[8])?null:new BigDecimal(values[8]));
                    bankPublicPrimitiveAum.setAmountMonthAverage(Objects.isNull(values[9])?null:new BigDecimal(values[9]));
                    bankPublicPrimitiveAum.setAmountLastMonthAverage(Objects.isNull(values[10])?null:new BigDecimal(values[10]));
                    bankPublicPrimitiveAum.setAmountQuarterAverage(Objects.isNull(values[11])?null:new BigDecimal(values[11]));
                    bankPublicPrimitiveAum.setAmountLastQuarterAverage(Objects.isNull(values[12])?null:new BigDecimal(values[12]));
                    bankPublicPrimitiveAum.setAmountYearAverage(Objects.isNull(values[13])?null:new BigDecimal(values[13]));
                    bankPublicPrimitiveAum.setAmountLastYearAverage(Objects.isNull(values[14])?null:new BigDecimal(values[14]));


                    bankPublicPrimitiveAum.setAmountRealMonthAverage(getRealAverage(bankPublicPrimitiveAum.getAmountMonthAverage(), daysOfMonth, allDaysOfMonth));
                    bankPublicPrimitiveAum.setAmountRealQuarterAverage(getRealAverage(bankPublicPrimitiveAum.getAmountQuarterAverage(), daysOfQuarter, allDaysOfQuarter));
                    bankPublicPrimitiveAum.setAmountRealYearAverage(getRealAverage(bankPublicPrimitiveAum.getAmountYearAverage(), daysOfYear, allDaysOfYear));

                    bankPublicPrimitiveAum.setUpdateTime(LocalDateTime.now());
                    batchCount++;
                    typeList.add(bankPublicPrimitiveAum);
                    batMap.put(values[2], typeList);
                    if (BasicConstant.INT_1000 == batchCount) {
                        bankPublicPrimitiveAumDao.saveBatchTypeSplit(batMap);
                        count += batchCount;
                        batchCount = 0;
                        batMap.clear();
                        log.debug("【对公客户AUM-m_ud_gxd_ent_cust_aum】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }
            } catch (Exception e) {
                rse.set(e);
                log.error("【对公客户AUM-m_ud_gxd_ent_cust_aum-error】" + Thread.currentThread().getName() + " batMap = {}", JsonUtils.objectToJson(batMap), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName), 40 * 1024);
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【对公客户AUM-m_ud_gxd_ent_cust_aum-error】队列插入异常");
                    throw new BaseException("【对公客户AUM-m_ud_gxd_ent_cust_aum 】队列插入异常");
                }
            }
            log.info("【对公客户AUM-m_ud_gxd_ent_cust_aum】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，得接数据

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【对公客户AUM-m_ud_gxd_ent_cust_aum】子进程尚未处理完毕");

                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【对公客户AUM-m_ud_gxd_ent_cust_aum 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【对公客户AUM-m_ud_gxd_ent_cust_aum 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【对公客户AUM-m_ud_gxd_ent_cust_aum】子线程全部完成");
        } catch (Exception e) {
            log.error("【对公客户AUM-m_ud_gxd_ent_cust_aum-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【对公客户AUM-m_ud_gxd_ent_cust_aum-error】银行数据读取错误");
        }
        log.info("【对公客户AUM-m_ud_gxd_ent_cust_aum】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【对公客户AUM-m_ud_gxd_ent_cust_aum-error】银行数据读取错误");
        }
    }

    /**
     * 客户经理信息表-BANK_SERVIC
     *
     * @param filePath 文件路径
     *//*
    @Override
    public void bankServicUpdate(String filePath) { *//*银行机构表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE	日	全量 	数据文件	M_UD_GXD_S_SRPB_ABL_BANK_SERVICE.unl*//*
        String fileName = "M_UD_GXD_S_SRPB_ABL_BANK_SERVICE.unl";
        log.info("【BankServicImpl bankServic】start====>{}", filePath + fileName); *//* 判断下文件是否存在*//*
        File dataFile = new File(filePath + fileName);
        if (!dataFile.exists()) {
            log.error("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE-error】M_UD_GXD_S_SRPB_ABL_BANK_SERVICE 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        } else { *//* 文件存在大小是否正常*//*
            if (dataFile.length() == 0) {
                log.error("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE-error】M_UD_GXD_S_SRPB_ABL_BANK_SERVICE 文件大小为0"); *//* throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);*//*
                return;
            }
        } *//* 先把数据清空,再插入*//*
        bankServicDao.truncate();
        try {
            List<BankServic> list = new ArrayList<>();
            BankServic bankServic;
            long count = 0L;
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                bankServic = new BankServic();
                String[] values = line.split("\u0001");
                if (values.length < 22) {
                    log.info("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE】 该条数据异常,需剔除 line = {}", line);
                    continue;
                }
                if (StringUtils.isBlank(values[0])) {
                    log.info("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE】 该条数据异常,需剔除 line = {}", line);
                    continue;
                }
                //1	ETL_DT	数据日期	VARCHAR(8)	N	1
                //2	ID	客户经理id	VARCHAR(20)	Y	2
                //3	SER_NAME	客户经理名称	VARCHAR(128)	N	3
                //4	LOGIN_NO	登录号	VARCHAR(20)	N	4
                //5	MOBILE	联系手机号	VARCHAR(20)	N	5
                //6	WORK_CARD_NO	工号	VARCHAR(20)	N	6
                //7	OWNER_ID	客户经理userid	VARCHAR(20)	N	7
                //8	IS_BANK_CLERK	是否是行内人员	VARCHAR(20)	N	8
                //9	DUTY_ID	职务id	VARCHAR(20)	N	9
                //10	BUSSI_TYPE	业务类型	VARCHAR(20)	N	10
                //11	CREATE_DATE	客户经理生效日期	VARCHAR(20)	N	11
                //12	SALES_IMG_URL	服务点图片	VARCHAR(128)	N	12
                //13	START_TIME	开始营业时间	VARCHAR(20)	N	13
                //14	END_TIME	结束营业时间	VARCHAR(20)	N	14
                //15	SLOGAN	宣传语	VARCHAR(200)	N	15
                //16	AREA_ID	区域id	VARCHAR(20)	N	16
                //17	ADDRESS	地址	VARCHAR(200)	N	17
                //18	STATUS	状态	VARCHAR(3)	N	18
                //19	DESCR	说明	VARCHAR(200)	N	19
                //20	ORG_ID	机构id	VARCHAR(20)	N	20
                //21	CREATE_TIME	创建时间	TIMESTAMP	N	21
                //22	UPDATE_TIME	更新时间	TIMESTAMP	N	22
                //23	QYKF_URL	企业微信客服链接	VARCHAR(100)	N	23
                //24	ORG_NO	机构编码	VARCHAR(20)	N	24
                {

                    bankServic.setId(values[1]);
                    bankServic.setSerName(values[2]);
                    bankServic.setLoginNo(values[3]);
                    bankServic.setMobile(values[4]);
                    bankServic.setWorkCardNo(values[5]);
                    bankServic.setOwnerId(values[6]);
                    bankServic.setIsBankClerk(values[7]);
                    bankServic.setDutyId(values[8]);
                    bankServic.setBussiType(values[9]);
                    bankServic.setCreateDate(LocalDateTimeUtil.of(cn.hutool.core.date.DateUtil.parse(values[10])));
                    bankServic.setSalesImgUrl(values[11]);
                    bankServic.setStartTime(LocalDateTimeUtil.of(cn.hutool.core.date.DateUtil.parse(values[12])));
                    bankServic.setEndTime(LocalDateTimeUtil.of(cn.hutool.core.date.DateUtil.parse(values[13])));
                    bankServic.setSlogan(values[14]);
                    bankServic.setAreaId(values[15]);
                    bankServic.setAddress(values[16]);
                    bankServic.setStatus(values[17]);
                    bankServic.setDescr(values[18]);
                    bankServic.setOrgId(values[19]);
                    bankServic.setCreateTime(LocalDateTimeUtil.of(cn.hutool.core.date.DateUtil.parse(values[20])));
                    bankServic.setUpdateTime(LocalDateTimeUtil.of(cn.hutool.core.date.DateUtil.parse(values[21])));
                    bankServic.setQykfUrl(values[22]);
                    bankServic.setOrgNo(values[23]);
                    bankServic.setEtlDt(values[0]);
                    *//* 最后一个字段是更新时间*//*
                    bankServic.setBankServicUpdateTime(LocalDateTime.now());
                    if(StringUtils.isBlank(bankServic.getId())){
                        log.info("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE id is null 】 该条数据异常,需剔除 line = {}", line);
                        continue;
                    }
                }

                list.add(bankServic);
                if (BasicConstant.INT_1000 == list.size()) {
                    bankServicDao.saveBatch(list);
                    count += list.size();
                    list.clear();
                    log.debug("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE】-" + LocalDateTime.now() + "已完成" + count);
                }
            }
            log.info("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE】文件数据读取完毕");

            if (list.size() > BasicConstant.INT_0) {
                bankServicDao.saveBatch(list);
                count += list.size();
                list.clear();
            }
            log.debug("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE】-" + LocalDateTime.now() + "最终已完成" + count);


        } catch (Exception e) {
            log.error("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE-error】银行数据读取错误");
        }
        log.info("【客户经理信息表-M_UD_GXD_S_SRPB_ABL_BANK_SERVICE】end====>{}", filePath + fileName);

    }*/
    /**
     * 商户交易统计表
     *
     * @param filePath 文件路径
     */
    @Override
    public void merchantTradeInfo(String filePath) {
        //商户交易统计表	日	全量		数据文件	m_ud_gxd_mrch_txn_stat.unl 数据是T-2
        String fileName = "m_ud_gxd_mrch_txn_stat.unl";
        log.info("【商户交易统计表】start====>{}", filePath + fileName);
        LocalDate dataDate = getDataDateFromFilePath(filePath).minusDays(1);
        int daysOfMonth = DateTimeUtil.daysOfMonth(dataDate);
        int allDaysOfMonth = DateTimeUtil.allDaysOfMonth(dataDate);
        // 判断下文件是否存在
        File dataFile = new File(filePath + fileName);
        if (!dataFile.exists()) {
            log.error("【商户交易统计表-error】m_ud_gxd_mrch_txn_stat 文件不存在");
            throw new BaseException(ErrorCodeEnums.BANK_DATA_NOT_EXISTS_ERROR);
        } else {
            // 文件存在大小是否正常
            if (dataFile.length() == 0) {
                log.error("【商户交易统计表-error】m_ud_gxd_mrch_txn_stat 文件大小为0");
                // throw new BaseException(ErrorCodeEnums.BANK_DATA_LENGTH_ZERO_ERROR);
                return;
            }
        }

        List<ConditionsName> conditionsNameByClassId = conditionsNameDao.findConditionsNameByClassIdAndSetStatus(BankFileTypeEnum.BANK_PUBLIC_TRADE_PRIMITIVE.getCode() + "");
        if (CollectionUtils.isEmpty(conditionsNameByClassId)) {
            return;
        }

        // 创建一个最大线程数为 @nThreads 的线程池
        List<Integer> dictIds = conditionsNameByClassId.stream().map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());
        Map<Integer, Map<String, String>> dictsMap = dictDetailDao.dictsSc2GxdMapByDictIds(dictIds);

        List<Integer> clearCondIds = conditionsNameByClassId.stream().filter(e -> 1 == e.getIsSplitTableField()).map(ConditionsName::getConditionsDictId).filter(Objects::nonNull).collect(Collectors.toList());

        for (Integer clearCondId : clearCondIds) {
            for (Map.Entry<String, String> entry : dictsMap.get(clearCondId).entrySet()) {
                Map<String, String> tableSuffix = DynamicTableNameHelper.getTableSuffix();
                tableSuffix.put("tbl_bank_public_trade_primitive", entry.getKey());
                DynamicTableNameHelper.setTableSuffix(tableSuffix);
                // 由于 m_ud_gxd_mrch_txn_stat 是全量数据，所以在接受数据前，将表清空
                bankPublicTradePrimitiveDao.truncate();
            }

        }
        // 由于 m_ud_gxd_mrch_txn_stat 是全量数据，所以在接受数据前，将表清空
        bankPublicTradePrimitiveDao.truncate();


        // 创建一个最大线程数为 @nThreads 的线程池
        ExecutorService executorService = newFixedThreadPool(nThreads);
        // 创建一个队列用于子线程共享数据
        ArrayBlockingQueue<String> dateLineQueue = new ArrayBlockingQueue<String>(queueSize);

        AtomicReference<Exception> rse = new AtomicReference<>();
        Thread cust = new Thread(() -> {
            String line;
            String splitField = null;
            List<BankPublicTradePrimitive> typeList = new ArrayList<>();
            Map<String, List<BankPublicTradePrimitive>> batMap = new CaseInsensitiveMap<>();
            BankPublicTradePrimitive bankPublicTradePrimitive;
            try {
                long batchCount = 0L;
                long count = 0L;
                while (true) {
                    line = dateLineQueue.poll(queuePullTimeout, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(line)) {
                        if (executorService.isShutdown()) {
                            log.debug("【商户交易统计表】" + Thread.currentThread().getName() + "- 处理完当前子线程剩余任务数{}后，结束当前子线程", batchCount);
                            if (batchCount > BasicConstant.INT_0) {
                                bankPublicTradePrimitiveDao.saveBatchTypeSplit(batMap);
                                count += batchCount;
                                batMap.clear();
                            }
                            log.debug("【商户交易统计表】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "最终已完成" + count);
                            break;
                        }
                        continue;
                    }

                    bankPublicTradePrimitive = new BankPublicTradePrimitive();
                    String[] values = line.split("\u0001");
                    boolean ignore = false;
                    for (int i = 0; i < values.length; i++) {
                        for (ConditionsName conditionsName : conditionsNameByClassId) {
                            if (conditionsName.getSetValueStatus() == BasicConstant.INT_0 && (i + 1) == conditionsName.getConditionsValueKeyIndex()) {
                                String value = values[i];
                                if (StringUtils.isBlank(value)) {
                                    if (ValueNullAllowEnum.NOT_ALLOW.getValue().equals(conditionsName.getValueNullAllow())) {
                                        ignore = true;
                                        log.warn("【商户交易统计表】" + Thread.currentThread().getName() + " 该条数据{}字段值为{}异常,需剔除 line = {}", conditionsName.getConditionsValueKey(), value, line);
                                    }
                                    continue;
                                }


                                // 根据type来拆分表
                                if (1 == conditionsName.getIsSplitTableField()) {
                                    if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                        if (dictsMap.get(conditionsName.getConditionsDictId()).containsKey(value)) {
                                            splitField = value;
                                            typeList = batMap.getOrDefault(splitField, new ArrayList<>());
                                        } else {
                                            splitField = "";
                                            typeList = batMap.getOrDefault(splitField, new ArrayList<>());

                                        }
                                        value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                    }
                                } else {
                                    if (!CollectionUtils.isEmpty(dictsMap.get(conditionsName.getConditionsDictId()))) {
                                        value = dictsMap.get(conditionsName.getConditionsDictId()).getOrDefault(value, value);
                                    }
                                }

                                setValue(bankPublicTradePrimitive, conditionsName.getConditionsValueKey(), value);
                            }
                        }
                    }
                    if (ignore) {
                        continue;
                    }
                    batchCount++;
                    bankPublicTradePrimitive.setUpdateTime(LocalDate.from(LocalDateTime.now()));
                    // 计算实际日均
                    bankPublicTradePrimitive.setMonthRealAverageAmount(getRealAverage(bankPublicTradePrimitive.getMonthAverageAmount(), daysOfMonth, allDaysOfMonth));
                    bankPublicTradePrimitive.setLastMonthRealAverageAdditionAmount(getAddition(bankPublicTradePrimitive.getMonthRealAverageAmount(), bankPublicTradePrimitive.getLastMonthAverageAmount()));

                    typeList.add(bankPublicTradePrimitive);
                    batMap.put(splitField, typeList);

                    if (BasicConstant.INT_1000 == batchCount) {
                        bankPublicTradePrimitiveDao.saveBatchTypeSplit(batMap);
                        count += batchCount;
                        batchCount = 0;
                        batMap.clear();
                        log.debug("【商户交易统计表】" + Thread.currentThread().getName() + "-" + LocalDateTime.now() + "已完成" + count);
                    }
                }
            } catch (Exception e) {
                rse.set(e);
                log.error("【商户交易统计表-error】" + Thread.currentThread().getName() + " batMap = {}", JsonUtils.objectToJson(batMap), e);
            }
        });

        for (int i = 0; i < nThreads - 1; i++) {
            executorService.submit(cust);
        }

        try {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath + fileName));
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                if (!dateLineQueue.offer(line, queueInsertTimeout, TimeUnit.MILLISECONDS)) {
                    dateLineQueue.clear();
                    executorService.shutdownNow();
                    log.error("【商户交易统计表-error】队列插入异常");
                    throw new BaseException("【商户交易统计表 】队列插入异常");
                }
            }
            log.info("【商户交易统计表】文件数据读取完毕");
            // todo 读取完毕后自己也不能闲着，可以继续处理数据，这里留着以后优化，现在足够用了

            int errNum = 0;
            while (!dateLineQueue.isEmpty()) {
                log.info("【商户交易统计表】子进程尚未处理完毕");
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【商户交易统计表 】dateLineQueue处理异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            errNum = 0;
            executorService.shutdown();
            while (!executorService.isTerminated()) {
                errNum++;
                if (errNum > maxErrNum) {
                    throw new BaseException("【商户交易统计表 】executorService shutdown异常");
                }
                Thread.sleep(spinLocWaitTime);
            }
            log.info("【商户交易统计表】子线程全部完成");
        } catch (Exception e) {
            log.error("【商户交易统计表-error】", e);
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【商户交易统计表-error】银行数据读取错误");
        }
        log.info("【商户交易统计表】end====>{}", filePath + fileName);
        if (Objects.nonNull(rse.get())) {
            throw new BaseException(ErrorCodeEnums.BANK_DATA_READ_IS_ERROR.getErrcode(), "【商户交易统计表-error】银行数据读取错误");
        }
    }

    public static void setValue(Object obj, String fieldName, String value) {

        // 创建一个对象
//        PrivatePrimitive myObject = new PrivatePrimitive();

        // 获取字段的Class对象
//        Class<?> clazz = myObject.getClass();
        Class<?> clazz = obj.getClass();
        // 通过字段名获取Field对象
        Field field = null;
        try {
            field = clazz.getDeclaredField(fieldName);

            // 如果字段是私有的，需要设置setAccessible(true)来绕过访问权限检查
            field.setAccessible(true);

            if (field.getType().equals(Integer.class)) {
                // 如果属性是int类型，而值是String类型，尝试转换

                Integer intValue = Integer.parseInt(value);
                field.set(obj, intValue);
            } else if (field.getType().equals(Long.class)) {
                Long intValue = Long.parseLong(value);
                field.set(obj, intValue);
            } else if (field.getType().equals(Double.class)) {
                Double doubleValue = Double.parseDouble(value);
                field.set(obj, doubleValue);
            } else if (field.getType().equals(BigDecimal.class)) {
                BigDecimal bigDecimalValue = new BigDecimal(value);
                field.set(obj, bigDecimalValue);
            } else if (field.getType().equals(LocalDateTime.class)) {

                DateTime parse = cn.hutool.core.date.DateUtil.parse(value);
                LocalDateTime localDate = LocalDateTimeUtil.of(parse);
                field.set(obj, localDate);
            } else if (field.getType().equals(LocalDate.class)) {
                LocalDate localDate = DateTimeUtil.parseYYYYMMDD(getYYYYMMDDDate(value));
                field.set(obj, localDate);
            } else {
                // 直接设置值，确保类型匹配
                field.set(obj, value);
            }
        } catch (NoSuchFieldException | IllegalAccessException e) {
            log.error("【ManageTagInfoServiceImpl.setValue赋值错误-dataerror {} {}】", fieldName, value, e);
            throw new BaseException(ErrorCodeEnums.BANK_FILE_SET_VALUE_ERROR);
        }


        // 给字段赋值
//        try {
//            field.set(obj, value);
//        } catch (IllegalAccessException e) {
//            e.printStackTrace();
//        }
    }

    public static String getYYYYMMDDDate(String date) {

//String result = date;

        String dateType = date.trim();

        dateType = dateType.replace("/", "").replace(".", "").replace("-", "").replace("年", "").replace("月", "").replace("日", "");

        return QxAndJhUtil.get8VaildDate(dateType);

    }

    private ExecutorService newFixedThreadPool(int n) {
        return new ThreadPoolExecutor(n, n,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(n * 2));
    }

    private BigDecimal getRealAverage(BigDecimal scAverage, int days, int zDays) {

        if (Objects.isNull(scAverage)) {
            return null;
        }

        return scAverage.multiply(BigDecimal.valueOf(days)).divide(BigDecimal.valueOf(zDays), 2, RoundingMode.HALF_UP);
    }

    private BigDecimal getAddition(BigDecimal decimal1, BigDecimal decimal2) {

        if (Objects.isNull(decimal1) || Objects.isNull(decimal2)) {
            return null;
        }

        return decimal1.subtract(decimal2);
    }

    private LocalDate getDataDateFromFilePath(String filePath) {
        // filePath 规则 localPath + zipFilePrefix + s + File.separator 要获取s
        filePath = filePath.replace(File.separator, "");
        String dataDateStr = filePath.substring(filePath.length() - 8);
        return DateTimeUtil.parseDateFormatStr(dataDateStr);
    }

}
